package com.lichao.FinkDataProcess.sink;

import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.codehaus.janino.ClassBodyEvaluator;

import java.util.HashMap;
import java.util.Map;

public class DynamicMysqlSinkFactory {

    // 动态生成并编译 Sink 类的源代码
    private static Class<?> compileSinkClass(String tableName) throws Exception {
        String className = "DynamicMysqlSink_" + tableName;
        String classSource = "\n" +
                "import org.apache.flink.configuration.Configuration;\n" +
                "import org.apache.flink.streaming.api.functions.sink.SinkFunction;\n" +
                "\n" +
                "import java.sql.Connection;\n" +
                "import java.sql.DriverManager;\n" +
                "import java.sql.PreparedStatement;\n" +
                "import java.util.Properties;\n" +
                "\n" +
                "public class " + className + " extends SinkFunction<String> {\n" +
                "\n" +
                "    private final String " + tableName + ";\n" +
                "\n" +
                "    public " + className + "(String " + tableName + ") {\n" +
                "        this." + tableName + " = " + tableName + ";\n" +
                "    }\n" +
                "\n" +
                "    PreparedStatement stmt;\n" +
                "\n" +
                "    public void open(Configuration parameters) throws Exception {\n" +
                "        super.open(parameters);\n" +
                "        Properties info = new Properties(); //定义Properties对象\n" +
                "        info.setProperty(\"user\", \"root\"); //设置Properties对象属性\n" +
                "        info.setProperty(\"password\", \"123456\");\n" +
                "        Class.forName(\"com.mysql.jdbc.Driver\"); //注册数据库驱动\n" +
                "        String url = \"jdbc:mysql://localhost:3306/test\"; //test为数据库名称\n" +
                "        Connection con = DriverManager.getConnection(url, info); //获取连接数据库的Connection对象\n" +
                "        stmt = con.prepareStatement(\"INSERT INTO \" + " + tableName + " + \" (url) VALUES (?)\");\n" +
                "    }\n" +
                "\n" +

                "    public void invoke(String value, Context context) throws Exception {\n" +
                "        super.invoke(value, context);\n" +
                "        stmt.setString(1, value);\n" +
                "        stmt.executeBatch();\n" +
                "    }\n" +
                "}";


        // 使用 Janino 编译源代码
        ClassBodyEvaluator evaluator = new ClassBodyEvaluator();
//        evaluator.setDefaultImports(new String[]{"org.apache.flink.streaming.api.functions.sink.SinkFunction"});
        evaluator.setClassName(className);
        evaluator.cook(classSource); // 这将编译并加载类
        return evaluator.getClazz();
    }

    // 根据表名创建 Sink 实例
    public static SinkFunction<String> createSink(String tableName) {
        try {
            Class<?> sinkClass = compileSinkClass(tableName);
            // 使用反射创建 Sink 实例
            return (SinkFunction<String>) sinkClass.getDeclaredConstructor().newInstance();
        } catch (Exception e) {
            throw new RuntimeException("Failed to create dynamic sink for table: " + tableName, e);
        }
    }

    public static void main(String[] args) {
        // 假设你有多个表名
        String[] tableNames = {"table1", "table2"};

        // 为每个表名创建 Sink 实例
        Map<String, SinkFunction<String>> sinks = new HashMap<>();
        for (String tableName : tableNames) {
            sinks.put(tableName, createSink(tableName));
        }

        // 现在你可以使用这些 sinks 来配置你的 Flink 作业了
        // ...

        // 这里只是打印出来以验证
        for (Map.Entry<String, SinkFunction<String>> entry : sinks.entrySet()) {
            System.out.println("Sink for table " + entry.getKey() + " created.");
            // 你可以在这里调用 entry.getValue().invoke(...) 来测试 Sink
        }
    }
}